package com.sec.moveFile.service;

import com.sec.moveFile.document.domain.SystemDocMoveDO;
import com.sec.moveFile.document.domain.SystemDocumentDO;
import com.sec.moveFile.document.repository.SystemDocMoveRepository;
import com.sec.moveFile.document.repository.SystemDocumentRepository;
import com.sec.moveFile.utils.NFSUtils;
import com.sec.moveFile.utils.S3Util;
import com.sec.moveFile.utils.specification.Specifications;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.core.io.Resource;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.*;

/**
 * @author 68300119
 * @Date 2021/5/12
 **/
@Service
public class MoveFileService {

    private static final Logger logger = LoggerFactory.getLogger(MoveFileService.class);


    private final SystemDocumentRepository systemDocumentRepository;

    private final SystemDocMoveRepository systemDocMoveRepository;

    private final S3Util s3Util;

    private final NFSUtils nfsUtils;

    private final AutowireCapableBeanFactory factory;

    public MoveFileService(SystemDocumentRepository systemDocumentRepository,
                           SystemDocMoveRepository systemDocMoveRepository,
                           S3Util s3Util,
                           NFSUtils nfsUtils,
                           AutowireCapableBeanFactory factory) {
        this.systemDocumentRepository = systemDocumentRepository;
        this.systemDocMoveRepository = systemDocMoveRepository;
        this.s3Util = s3Util;
        this.nfsUtils = nfsUtils;
        this.factory = factory;
    }


    /**
     * 迁移10万个文件
     *
     * @return 是否启动完成
     */
    public Boolean moveFile() throws InterruptedException {
        return this.moveFile(100000);
    }

    /**
     * 文件迁移
     *
     * @param moveSize 迁移数量
     * @return 是否启动完成
     */
    public Boolean moveFile(Integer moveSize) throws InterruptedException {
        logger.info("开始迁移");
        Specification<SystemDocumentDO> spec = Specifications.<SystemDocumentDO>and()
                .isNull("moveFlag")
                .notNull("url")
                .build();


        // 找出没有迁移过的前10W条数据
        Pageable page = PageRequest.of(0, moveSize, Sort.by("lastModifiedDate").descending());

        List<SystemDocumentDO> documents = systemDocumentRepository.findAll(spec, page).getContent();

        logger.info("迁移数据量：{}", documents.size());
        return this.move(documents);
    }

    /**
     * 迁移文件
     *
     * @param documents 文件数据
     * @return true
     * @throws InterruptedException
     */
    private boolean move(List<SystemDocumentDO> documents) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(documents.size());
        int poolSize = Runtime.getRuntime().availableProcessors() * 2;
        ThreadPoolExecutor pool = new ThreadPoolExecutor(poolSize,
                poolSize,
                0,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        documents.forEach(doc -> {
            try {
                pool.execute(() -> {
                    this.copyFile(doc.getId(), doc.getUrl());
                });
            } finally {
                countDownLatch.countDown();
            }
        });
        countDownLatch.await();
        pool.shutdown();
        logger.info("已全部完成！");
        return true;
    }

    /**
     * 从NFS复制文件到S3
     *
     * @param docId doc id
     * @param url   url
     */
    private void copyFile(String docId, String url) {
        SystemDocMoveDO systemDocMoveDO = new SystemDocMoveDO();
        systemDocMoveDO.setDocId(docId);
        try {
            // 从nfs下载
            Resource download = nfsUtils.download(url);
            // 上传到s3
            s3Util.upload(download, url);
            // 成功保存到迁移表
            systemDocMoveDO.setMoveFlag(true);
            systemDocMoveDO.setLog("完成");
        } catch (Exception e) {
            // 失败也保存到迁移表
            systemDocMoveDO.setMoveFlag(false);
            systemDocMoveDO.setLog(e.getMessage());
        } finally {
            systemDocMoveRepository.saveAndFlush(systemDocMoveDO);
        }
    }

    /**
     * 文件校验
     *
     * @return 是否启动完成
     */
    public Boolean check() throws InterruptedException {
        logger.info("开始校验");
        Specification<SystemDocumentDO> spec = Specifications.<SystemDocumentDO>and()
                .isNull("checkFlag")
                .notNull("url")
                .build();

        // 找出没有迁移过的前10W条数据
        List<SystemDocumentDO> documents = systemDocumentRepository.findAll(spec);

        logger.info("迁移数据量：{}", documents.size());
        return this.check(documents);
    }

    /**
     * 文件校验
     *
     * @param documents documents
     * @return true
     * @throws InterruptedException
     */
    private boolean check(List<SystemDocumentDO> documents) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(documents.size());
        int poolSize = Runtime.getRuntime().availableProcessors() * 2;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(poolSize,
                poolSize,
                0,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());
        documents.forEach(doc -> {
            executor.execute(() -> {
                this.checkFile(doc.getId(), doc.getUrl());
                countDownLatch.countDown();
            });
        });

        countDownLatch.await();
        executor.shutdown();
        logger.info("已全部完成！");
        return true;
    }

    /**
     * 检查文件
     *
     * @param docId doc id
     * @param url   url
     */
    private void checkFile(String docId, String url) {
        SystemDocMoveDO systemDocMoveDO = systemDocMoveRepository
                .findByDocId(docId)
                .orElseThrow(() -> new RuntimeException("没找到docId:" + docId));
        try {
            // 校验头
            s3Util.getHead(url);
            // 成功保存到迁移表
            systemDocMoveDO.setCheckFlag(true);
            systemDocMoveDO.setCheckLog("完成");
        } catch (Exception e) {
            // 失败也保存到迁移表
            systemDocMoveDO.setCheckFlag(false);
            systemDocMoveDO.setCheckLog(e.getMessage());
        } finally {
            systemDocMoveRepository.saveAndFlush(systemDocMoveDO);
        }
    }
}
